package com.abel.bigwater.msg

import org.apache.kafka.clients.producer.Partitioner
import org.apache.kafka.common.Cluster

class OctPartitioner : Partitioner {
    companion object {
        const val CNT_PARTITION = 7
    }

    /**
     * Configure this class with the given key-value pairs
     */
    override fun configure(configs: MutableMap<String, *>?) {
    }

    /**
     * This is called when partitioner is closed.
     */
    override fun close() {
    }

    /**
     * Compute the partition for the given record.
     *
     * @param topic The topic name
     * @param key The key to partition on (or null if no key)
     * @param keyBytes The serialized key to partition on( or null if no key)
     * @param value The value to partition on or null
     * @param valueBytes The serialized value to partition on or null
     * @param cluster The current cluster metadata
     */
    override fun partition(topic: String?, key: Any?, keyBytes: ByteArray?, value: Any?, valueBytes: ByteArray?, cluster: Cluster?): Int {
        if (key is Number) {
            return key.toInt().rem(CNT_PARTITION)
        } else if (key?.toString()?.toIntOrNull() ?: 0 > 0) {
            return key.toString().toInt().rem(CNT_PARTITION)
        } else {
            return CNT_PARTITION
        }
    }
}